package paxos

//
// Paxos library, to be included in an application.
// Multiple applications will run, each including
// a Paxos peer.
//
// Manages a sequence of agreed-on values.
// The set of peers is fixed.
// Copes with network failures (partition, msg loss, &c).
// Does not store anything persistently, so cannot handle crash+restart.
//
// The application interface:
//
// px = paxos.Make(peers []string, me string)
// px.Start(seq int, v interface{}) -- start agreement on new instance
// px.Status(seq int) (Fate, v interface{}) -- get info about an instance
// px.Done(seq int) -- ok to forget all instances <= seq
// px.Max() int -- highest instance seq known, or -1
// px.Min() int -- instances before this seq have been forgotten
//

import "net"
import "net/rpc"
import "log"

import "os"
import "syscall"
import "sync"
import "sync/atomic"
import "fmt"
import (
	"encoding/gob"
	"math/rand"
	"time"
//	"sort"
)

// px.Status() return values, indicating
// whether an agreement has been decided,
// or Paxos has not yet reached agreement,
// or it was agreed but forgotten (i.e. < Min()).
type Fate int

const (
	Decided   Fate = iota + 1
	Pending        // not yet decided.
	Forgotten      // decided but forgotten.
)

var Debug = 1

func DPrintf(format string, a ...interface{}) (n int, err error) {
	if Debug > 0 {
		log.Printf(format, a...)
	}
	return
}

type Paxos struct {
	mu         sync.Mutex
	l          net.Listener
	dead       int32 // for testing
	unreliable int32 // for testing
	rpcCount   int32 // for testing
	peers      []string
	me         int // index into peers[]

	// Your data here.

	instances  map[int]*Instance
	min        []int
	majority   int
	rwMu       sync.RWMutex

	check      func(interface{}, interface{}) bool
	handle     func(interface{})
	LastSeq    int
}

type Instance struct {
	Status     Fate
	ProposeNum int64
	AcceptNum  int64
	Value      interface{}
}

type ProposeArgs struct {
	Num int64
	Seq int
}

type ProposeReply struct {
	Ok        bool
	Value     interface{}
	AcceptNum int64
}

type AcceptArgs struct {
	Seq   int
	Num   int64
	Value interface{}
}

type AcceptReply struct {
	Ok bool
}

type DecideArgs struct {
	Seq   int
	Num   int64
	Value interface{}
	Min   int
	From  int
}

type DecideReply struct {
	Min int
}

func (px *Paxos) getInstance(seq int) (instance *Instance) {
	var ok bool
	if instance, ok = px.instances[seq]; !ok {
		instance = &Instance{Status: Pending, Value: nil}
		px.instances[seq] = instance
	}
	return
}

func (px *Paxos) Propose(args *ProposeArgs, reply *ProposeReply) error {
	px.rwMu.Lock()
	defer px.rwMu.Unlock()

	instance := px.getInstance(args.Seq)
	if args.Num > instance.ProposeNum {
		instance.ProposeNum = args.Num

		reply.Ok = true
		reply.AcceptNum = instance.AcceptNum
		reply.Value = instance.Value
	} else {
		reply.Ok = false
	}
	return nil
}

func (px *Paxos) Accept(args *AcceptArgs, reply *AcceptReply) error {
	px.rwMu.Lock()
	defer px.rwMu.Unlock()

	instance := px.getInstance(args.Seq)
	if args.Num >= instance.ProposeNum {
		instance.ProposeNum = args.Num
		instance.AcceptNum = args.Num
		instance.Value = args.Value

		reply.Ok = true
	} else {
		reply.Ok = false
	}
	return nil
}

func (px *Paxos) Decide(args *DecideArgs, reply *DecideReply) error {
	px.rwMu.Lock()

	instance := px.getInstance(args.Seq)
	instance.ProposeNum = args.Num
	instance.AcceptNum = args.Num
	instance.Value = args.Value
	instance.Status = Decided
	px.min[args.From] = args.Min
	px.rwMu.Unlock()

	reply.Min = px.min[px.me]
	return nil
}

//
// call() sends an RPC to the rpcname handler on server srv
// with arguments args, waits for the reply, and leaves the
// reply in reply. the reply argument should be a pointer
// to a reply structure.
//
// the return value is true if the server responded, and false
// if call() was not able to contact the server. in particular,
// the replys contents are only valid if call() returned true.
//
// you should assume that call() will time out and return an
// error after a while if it does not get a reply from the server.
//
// please use call() to send all RPCs, in client.go and server.go.
// please do not change this function.
//
func call(srv string, name string, args interface{}, reply interface{}) bool {
	c, err := rpc.Dial("unix", srv)
	if err != nil {
		err1 := err.(*net.OpError)
		if err1.Err != syscall.ENOENT && err1.Err != syscall.ECONNREFUSED {
			//fmt.Printf("paxos Dial() failed: %v\n", err1)
		}
		return false
	}
	defer c.Close()

	err = c.Call(name, args, reply)
	if err == nil {
		return true
	}

	//fmt.Println(err)
	return false
}

func (px *Paxos) Update() {
	for {
		status, obj := px.Status(px.LastSeq)
		switch status {
		case Decided:
			px.handle(obj)
			px.Done(px.LastSeq)
			px.LastSeq += 1
		case Forgotten:
			return
		case Pending:
			return
		}
	}
}

func (px *Paxos) MakePropose(value interface{}) {
	for {
		var decidedValue interface{}

		px.Start(px.LastSeq, value)
		//
		//if px.me == 0 {
		//	array := make([]int, 0)
		//	for k, _ := range px.instances  {
		//		array = append(array, k)
		//	}
		//	sort.Ints(array)
		//	fmt.Println(px.LastSeq, array)
		//}

		to := time.Millisecond * 10
		for {
			status, obj := px.Status(px.LastSeq)
			if status == Decided {
				decidedValue = obj
				break
			} else {
				if to < time.Second*10 {
					to *= 2
				}
				time.Sleep(to)
			}
		}
		px.handle(decidedValue)
		px.Done(px.LastSeq)
		px.LastSeq += 1

		if px.check(decidedValue, value) {
			break
		}
	}
}

func (px *Paxos) Save(enc *gob.Encoder) {
	//px.rwMu.RLock()
	//defer px.rwMu.RUnlock()

	enc.Encode(px.instances)
	px.Min()
	enc.Encode(px.min)
	enc.Encode(px.LastSeq)
}

func (px *Paxos) Resume(dec *gob.Decoder) {
	px.rwMu.Lock()
	defer px.rwMu.Unlock()

	dec.Decode(&px.instances)
	dec.Decode(&px.min)
	dec.Decode(&px.LastSeq)
}

//
// the application wants paxos to start agreement on
// instance seq, with proposed value v.
// Start() returns right away; the application will
// call Status() to find out if/when agreement
// is reached.
//
func (px *Paxos) Start(seq int, v interface{}) {
	// Your code here.

	if seq < px.Min() {
		return
	}

	go func() {
		count := 0
		for !px.isdead() {
			count += 1
			if status, _ := px.Status(seq); status == Decided {
				break
			}
			var value interface{} = nil
			num := time.Now().UnixNano()
			{
				args := &ProposeArgs{Num: num, Seq: seq}
				count := 0
				var maxAccept int64 = 0

				for i, peer := range px.peers {
					if count >= px.majority {
						break
					}
					reply := &ProposeReply{}
					if i == px.me {
						px.Propose(args, reply)
					} else {
						call(peer, "Paxos.Propose", args, reply)
					}

					if reply.Ok {
						count += 1
						if reply.AcceptNum > maxAccept {
							maxAccept = reply.AcceptNum
							value = reply.Value
						}
					}
				}
				if count < px.majority {
					continue
				}

				if value == nil {
					value = v
				}
			}

			{
				args := &AcceptArgs{Num: num, Seq: seq, Value: value}
				count := 0

				for i, peer := range px.peers {
					if count >= px.majority {
						break
					}
					reply := &AcceptReply{}

					if i == px.me {
						px.Accept(args, reply)
					} else {
						call(peer, "Paxos.Accept", args, reply)
					}

					if reply.Ok {
						count += 1
					}
				}

				if count < px.majority {
					continue
				}

			}

			{
				px.rwMu.RLock()
				args := &DecideArgs{Num: num, Seq: seq, Value: value, Min: px.min[px.me], From: px.me}
				px.rwMu.RUnlock()

				for i, peer := range px.peers {
					reply := &DecideReply{}

					if i == px.me {
						px.Decide(args, reply)
					} else {
						if ok := call(peer, "Paxos.Decide", args, reply);ok {
							if reply.Min > px.min[i] {
								px.min[i] = reply.Min
							}
						}
					}
				}
			}
			break
		}
	}()
}

//
// the application on this machine is done with
// all instances <= seq.
//
// see the comments for Min() for more explanation.
//
func (px *Paxos) Done(seq int) {
	// Your code here.
	px.rwMu.Lock()
	px.min[px.me] = seq
	px.rwMu.Unlock()
}

//
// the application wants to know the
// highest instance sequence known to
// this peer.
//
func (px *Paxos) Max() int {
	// Your code here.
	px.rwMu.RLock()
	max := 0
	for k, _ := range px.instances {
		if k > max {
			max = k
		}
	}
	px.rwMu.RUnlock()
	return max
}

//
// Min() should return one more than the minimum among z_i,
// where z_i is the highest number ever passed
// to Done() on peer i. A peers z_i is -1 if it has
// never called Done().
//
// Paxos is required to have forgotten all information
// about any instances it knows that are < Min().
// The point is to free up memory in long-running
// Paxos-based servers.
//
// Paxos peers need to exchange their highest Done()
// arguments in order to implement Min(). These
// exchanges can be piggybacked on ordinary Paxos
// agreement protocol messages, so it is OK if one
// peers Min does not reflect another Peers Done()
// until after the next instance is agreed to.
//
// The fact that Min() is defined as a minimum over
// *all* Paxos peers means that Min() cannot increase until
// all peers have been heard from. So if a peer is dead
// or unreachable, other peers Min()s will not increase
// even if all reachable peers call Done. The reason for
// this is that when the unreachable peer comes back to
// life, it will need to catch up on instances that it
// missed -- the other peers therefor cannot forget these
// instances.
//
func (px *Paxos) Min() int {
	// You code here.
	px.rwMu.Lock()
	defer px.rwMu.Unlock()

	min := px.min[px.me]
	for _, v := range px.min {
		if v < min {
			min = v
		}
	}

	for k, _ := range px.instances {
		if k < min {
			delete(px.instances, k)
		}
	}
	return min + 1
}

//
// the application wants to know whether this
// peer thinks an instance has been decided,
// and if so what the agreed value is. Status()
// should just inspect the local peer state;
// it should not contact other Paxos peers.
//
func (px *Paxos) Status(seq int) (Fate, interface{}) {
	// Your code here.

	px.rwMu.RLock()
	defer px.rwMu.RUnlock()
	if instance, ok := px.instances[seq]; ok {
		return instance.Status, instance.Value
	} else {
		return Forgotten, nil
	}
}

func (px *Paxos) SetFunc(check func(interface{}, interface{}) bool, handle func(interface{})) {
	px.check = check
	px.handle = handle
}

//
// tell the peer to shut itself down.
// for testing.
// please do not change these two functions.
//
func (px *Paxos) Kill() {
	atomic.StoreInt32(&px.dead, 1)
	if px.l != nil {
		px.l.Close()
	}
}

//
// has this peer been asked to shut down?
//
func (px *Paxos) isdead() bool {
	return atomic.LoadInt32(&px.dead) != 0
}

// please do not change these two functions.
func (px *Paxos) setunreliable(what bool) {
	if what {
		atomic.StoreInt32(&px.unreliable, 1)
	} else {
		atomic.StoreInt32(&px.unreliable, 0)
	}
}

func (px *Paxos) isunreliable() bool {
	return atomic.LoadInt32(&px.unreliable) != 0
}

//
// the application wants to create a paxos peer.
// the ports of all the paxos peers (including this one)
// are in peers[]. this servers port is peers[me].
//
func Make(peers []string, me int, rpcs *rpc.Server) *Paxos {
	px := &Paxos{}
	px.peers = peers
	px.me = me

	// Your initialization code here.

	px.instances = make(map[int]*Instance)
	px.min = make([]int, len(peers))
	px.majority = len(peers)/2 + 1
	px.rwMu = sync.RWMutex{}
	px.mu = sync.Mutex{}

	for k, _ := range peers {
		px.min[k] = -1
	}

	if rpcs != nil {
		// caller will create socket &c
		rpcs.Register(px)
	} else {
		rpcs = rpc.NewServer()
		rpcs.Register(px)

		// prepare to receive connections from clients.
		// change "unix" to "tcp" to use over a network.
		os.Remove(peers[me]) // only needed for "unix"
		l, e := net.Listen("unix", peers[me])
		if e != nil {
			log.Fatal("listen error: ", e)
		}
		px.l = l

		// please do not change any of the following code,
		// or do anything to subvert it.

		// create a thread to accept RPC connections
		go func() {
			for px.isdead() == false {
				conn, err := px.l.Accept()
				if err == nil && px.isdead() == false {
					if px.isunreliable() && (rand.Int63()%1000) < 100 {
						// discard the request.
						conn.Close()
					} else if px.isunreliable() && (rand.Int63()%1000) < 200 {
						// process the request but force discard of reply.
						c1 := conn.(*net.UnixConn)
						f, _ := c1.File()
						err := syscall.Shutdown(int(f.Fd()), syscall.SHUT_WR)
						if err != nil {
							fmt.Printf("shutdown: %v\n", err)
						}
						atomic.AddInt32(&px.rpcCount, 1)
						go rpcs.ServeConn(conn)
					} else {
						atomic.AddInt32(&px.rpcCount, 1)
						go rpcs.ServeConn(conn)
					}
				} else if err == nil {
					conn.Close()
				}
				if err != nil && px.isdead() == false {
					fmt.Printf("Paxos(%v) accept: %v\n", me, err.Error())
				}
			}
		}()
	}

	return px
}
